package com.audaque.springboot.foshanupload.rocketmq.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Objects;


@Slf4j
@Component
@RocketMQTransactionListener
public class LocalTransactionListener implements RocketMQLocalTransactionListener {

    // 生产者数据库事务逻辑写在这里
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
        MessageHeaders headers = message.getHeaders();
        // 本地事务id
        String transactionId = (String) headers.get(RocketMQHeaders.PREFIX + RocketMQHeaders.TRANSACTION_ID);

        String topic = (String) headers.get("topic");
        if (!StringUtils.hasLength(topic)) {
            // 直接回滚
            return RocketMQLocalTransactionState.ROLLBACK;
        }

        if (Objects.equals(topic, "transaction_topic:tx")) {
            String orderId = (String) headers.get("orderId");
            if (!StringUtils.hasLength(orderId)) {
                // 直接回滚
                return RocketMQLocalTransactionState.ROLLBACK;
            }
            try {
                int id = Integer.parseInt(orderId);

                // 执行保存 orderService.saveOrder(order) 执行本地事务


                // 模拟本地事务执行成功(偶数)和失败(奇数)
                if (id % 2 == 0) {
                    // 本地事务成功则插入本地事务日志表：本地事务id
                    return RocketMQLocalTransactionState.COMMIT;
                } else {
                    // 假设这里的失败是本地事务还在执行(还不确定提交还是回滚)
                    return RocketMQLocalTransactionState.UNKNOWN;
                }
            } catch (Exception e) {
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }

    /**
     * 检查事务状态（监听回查请求）
     * 1.当成产者执行本地事务发生故障或者是返回 UNKNOWN 状态,要保证这条消息最终被消费,RocketMQ会像服务端发送回查请求,确认本地事务的执行状态
     * 2.不会无休止的的信息事务状态回查，默认回查15次，如果15次回查还是无法得知事务状态，RocketMQ默认回滚该消息
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        MessageHeaders headers = message.getHeaders();
        String transactionId = (String) headers.get(RocketMQHeaders.PREFIX + RocketMQHeaders.TRANSACTION_ID);
        String orderId = (String) headers.get("orderId");
        log.debug("check transaction start transactionId = {} [orderId = {}]", transactionId, orderId);
        if (!StringUtils.hasLength(orderId)) {
            // 直接回滚
            return RocketMQLocalTransactionState.ROLLBACK;
        }

        // 查询 orderService.getOrderById(id)或者查本地事务日志表：本地事务id
        // 如果能查到则说明本地事务执行成功 返回rocketMQLocalTransactionState.COMMIT
        // 反之则说明本地事务还在执行或者是出现故障

        // 这里统一模拟为出现故障，不让消费者消费消息
        return RocketMQLocalTransactionState.ROLLBACK;
    }


}
